package distributedcounter;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.HashSet;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

/**
 * @author Yinzi Chen
 * @date Jan 27, 2014
 */
public class MultiThreadCounter implements Runnable {
    private int diskId;
    private String fileDir;
    private static HashSet<String>[] strSet;

    public MultiThreadCounter(int diskId, String fileDir) {
        this.diskId = diskId;
        this.fileDir = fileDir;
        strSet[diskId] = new HashSet<String>();
    }

    public void run() {
        try {
            File dir = new File(fileDir);
            File[] textFiles = dir.listFiles();
            BufferedReader br;
            String line;
            for (File textFile : textFiles) {
                br = new BufferedReader(new InputStreamReader(
                        new FileInputStream(textFile)));
                while ((line = br.readLine()) != null) {
                    strSet[diskId].add(line);
                }
                br.close();
            }
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        int diskNum = 12;
        String fileDir = "/data/";
        strSet = new HashSet[12];
        if (args.length > 0) {
            diskNum = Integer.parseInt(args[0]);
        }
        if (args.length > 1) {
            fileDir = args[1];
        }
        ExecutorService service = Executors.newFixedThreadPool(diskNum);
        Future[] futures = new Future[diskNum];
        for (int i = 0; i < diskNum; ++i) {
            futures[i] = service.submit(new MultiThreadCounter(i, i + "/"
                    + fileDir));
        }
        for (int i = 0; i < diskNum; ++i) {
            try {
                futures[i].get();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }
        long totLines = 0;
        for (HashSet<String> set : strSet) {
            totLines += set.size();
        }
        System.out.println(totLines);
    }
}
